package cn.lsh.hbase.coprocessor;

import cn.lsh.protobuf.Sum;
import com.google.protobuf.RpcCallback;
import com.google.protobuf.RpcController;
import com.google.protobuf.Service;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.coprocessor.CoprocessorException;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.RegionScanner;
import org.apache.hadoop.hbase.shaded.protobuf.ResponseConverter;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

/**
 * 自定义的一个求和类
 */
public class SumEndPoint extends Sum.SumService implements RegionCoprocessor {
	Logger log = LoggerFactory.getLogger(SumEndPoint.class);

	private RegionCoprocessorEnvironment env;

	@Override
	public void getSum(RpcController controller, Sum.SumRequest request, RpcCallback<Sum.SumResponse> done) {
		//求和
		Scan scan = new Scan();
		scan.addFamily(Bytes.toBytes(request.getFamily()));
		scan.addColumn(Bytes.toBytes(request.getFamily()), Bytes.toBytes(request.getColumn()));
		RegionScanner scanner = null;
		Sum.SumResponse response = null;
		try {
			List<Cell> results = new ArrayList<>();
			boolean hasMore;
			log.info("renv: " + env);
			log.info("regionInfo: " + env.getRegionInfo());
			log.info("region:" + env.getRegion());
			scanner = env.getRegion().getScanner(scan);
			long sum = 0L;
			do {
				hasMore = scanner.next(results);
				for (Cell cell : results) {
					sum += Bytes.toLong(CellUtil.cloneValue(cell));
				}
				results.clear();
			} while (hasMore);
			response = Sum.SumResponse.newBuilder().setSum(sum).build();
		} catch (IOException e) {
			ResponseConverter.setControllerException(controller, e);
		} finally {
			if (scanner != null) {
				try {
					scanner.close();
				} catch (IOException e) {
				}
			}
		}
		done.run(response);
	}

	@Override
	public void start(CoprocessorEnvironment env) throws IOException {
		log.info("init env: " + env);
		if (env instanceof RegionCoprocessorEnvironment) {
			log.info("init env");
			this.env = (RegionCoprocessorEnvironment) env;
		} else {
			throw new CoprocessorException("Must be loaded on a table region");
		}
	}

	@Override
	public void stop(CoprocessorEnvironment env) throws IOException {
		//求和计算结束
		log.info("stop env: " + env);
	}

	@Override
	public Iterable<Service> getServices() {
		log.info("getServices");
		return Collections.singletonList(this);
	}
}
